package com.xes.rabbitmq.job.handler;

import com.alibaba.fastjson.JSON;
import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.xes.rabbitmq.core.constant.MessageConstant;
import com.xes.rabbitmq.core.domain.tables.daos.TbMessageDao;
import com.xes.rabbitmq.core.domain.tables.pojos.TbMessage;
import com.xes.rabbitmq.core.domain.tables.pojos.TbStudent;
import com.xes.rabbitmq.core.enums.MessageStatusEunm;
import com.xes.rabbitmq.core.service.SenderStudentService;
import lombok.extern.slf4j.Slf4j;
import org.jooq.DSLContext;
import org.jooq.Record;
import org.jooq.SelectConditionStep;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;

import static com.xes.rabbitmq.core.domain.Tables.TB_MESSAGE;

@Slf4j
@Component
public class MessageSenderTask implements SimpleJob {

    @Autowired
    private DSLContext dsl;

    @Autowired
    private TbMessageDao messageDao;

    @Autowired
    private SenderStudentService senderStudentService;

    @Override
    public void execute(ShardingContext shardingContext) {

        SelectConditionStep<Record> where = dsl.select().from(TB_MESSAGE)
                .where(TB_MESSAGE.STATUS.eq(MessageStatusEunm.SENDING.getValue())
                        .and(TB_MESSAGE.NEXT_RETRY_TIME.le(LocalDateTime.now())));

        log.info("sql={}", where.getSQL());

        List<TbMessage> messages = where.fetchInto(TbMessage.class);

        log.info("records={}", JSON.toJSONString(messages));

        messages.forEach(message -> {
            message.setRetryTimes(message.getRetryTimes() + 1);
            if (message.getRetryTimes() < MessageConstant.MAX_RETRY_TIMES) {
                log.info(JSON.toJSONString(message));
                LocalDateTime nextRetryTime = this.calcuNextTime(message.getRetryTimes());
                int execute = dsl.update(TB_MESSAGE).set(TB_MESSAGE.NEXT_RETRY_TIME, nextRetryTime)
                        .set(TB_MESSAGE.RETRY_TIMES, TB_MESSAGE.RETRY_TIMES.add(1))
                        .set(TB_MESSAGE.OPERATE_TIME, LocalDateTime.now())
                        .where(TB_MESSAGE.ID.eq(message.getId()))
                        .execute();
                log.info("影响行数：{}", execute);

                TbStudent student = JSON.parseObject(message.getMessage(), TbStudent.class);
                senderStudentService.send(student, message.getId());
            } else {
                message.setStatus(MessageStatusEunm.FAIL.getValue());
                messageDao.update(message);
            }

        });


    }


    /**
     * 计算下次重试时间
     *
     * @param retryTimes
     * @return
     */
    private LocalDateTime calcuNextTime(Integer retryTimes) {
        Integer nextRetryTime = 0;
        if (Objects.equals(retryTimes, MessageConstant.RETRY_TIMES_1)) {
            nextRetryTime = MessageConstant.TIMEOUT_1;
        } else if (Objects.equals(retryTimes, MessageConstant.RETRY_TIMES_2)) {
            nextRetryTime = MessageConstant.TIMEOUT_2;
        } else if (Objects.equals(retryTimes, MessageConstant.RETRY_TIMES_3)) {
            nextRetryTime = MessageConstant.TIMEOUT_3;
        }
        return LocalDateTime.now().plusMinutes(nextRetryTime);
    }
}
